package com.wudl.flink.stream.sink

import com.wudl.flink.stream.source.TransformDemo.SensorReading
import org.apache.flink.api.common.serialization.SimpleStringEncoder
import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._

object FileSink {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val inputStream = env.readTextFile("F:\\ideaWorkSpace2020\\demo\\Flink-wudl\\src\\main\\resources\\sensor.txt")
    val dataStream = inputStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    }
    )
    dataStream.print()

    dataStream.addSink(StreamingFileSink.forRowFormat(new Path("F:\\ideaWorkSpace2020\\demo\\Flink-wudl\\src\\main\\resources"), new SimpleStringEncoder[SensorReading]()).build())
    env.execute()

  }
}
